package org.flk;

import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

public class SqlSeverSource extends RichSourceFunction<Map<String,Object>> {
    private Connection connection=null;
    private String beginTime=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
        String url = "jdbc:sqlserver://127.0.0.1:1433;databaseName = sdc";
        connection = DriverManager.getConnection(url,"sa","sa123456");
        ExecutionConfig.GlobalJobParameters globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        Configuration globConf = (Configuration) globalParams;
        //beginTime = globConf.getString("beginTime", null);
        beginTime="2022-12-10 00:00:00";
    }

    @Override
    public void run(SourceContext<Map<String, Object>> sourceContext) throws Exception {
        PreparedStatement ps = null;
        while (true) {
            Date bgd=DateUtils.parseDate(beginTime,"yyyy-MM-dd HH:mm:ss");
            Date bgd2= DateUtils.addHours(bgd,1);
            if(new Date().before(bgd2)){
                continue;
            }
            DateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String endTime=dateformat.format(bgd2);
            ps = (PreparedStatement) connection
                    .prepareStatement(
                            "SELECT TOP(1000) [id]\n" +
                                    "      ,[msg]\n" +
                                    "      ,[save_time]\n" +
                                    "      ,[ip]\n" +
                                    "      ,[sr_ip]\n" +
                                    "      ,[sr_port]\n" +
                                    "      ,[port]\n" +
                                    "      ,[tag_ip]\n" +
                                    "      ,[tag_port]\n" +
                                    "  FROM [sdc].[dbo].[syslog] where [save_time] between '" + beginTime
                    +"' and '"+endTime+"'");
            ResultSet resultSet = ps.executeQuery();
            while (resultSet.next()) {
                Map<String, Object> map = new HashMap<>();
                map.put("id", resultSet.getString("id"));
                map.put("msg", resultSet.getString("msg"));
                map.put("saveTime", resultSet.getString("save_time"));
                map.put("ip", resultSet.getString("ip"));
                sourceContext.collect(map);
            }

            beginTime = dateformat.format(bgd2);

        }

    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        connection.close();
        super.close();
    }
}
